package com.henry.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 并发消费者
 *
 * @author Liu Hanlin
 * @create 2025-11-10 23:17
 */
@Slf4j
public abstract class ConcurrentConsumerHandler extends ConcurrentConsumer {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context){

        for (MessageExt message : messages) {

            try {
                String tags = message.getTags();
                String topic = message.getTopic();
                String msgId = message.getMsgId();
                String keys = message.getKeys();
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);

                log.info("收到消息, topic:[{}], tags:[{}], msgId:[{}], keys:[{}], msg:[{}]", topic, tags, msgId, keys, msg);
                this.consumeMessage(msgId, keys, msg);
            } catch (Exception e) {
                log.error("消息处理失败, msgId:[{}]", message.getMsgId(), e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

    }


}
